package com.dlg.exchange.engin;

import com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure;
import com.alipay.remoting.exception.CodecException;
import com.alipay.sofa.jraft.rhea.client.DefaultRheaKVStore;
import com.alipay.sofa.jraft.rhea.client.RheaKVStore;
import com.alipay.sofa.jraft.rhea.options.PlacementDriverOptions;
import com.alipay.sofa.jraft.rhea.options.RegionRouteTableOptions;
import com.alipay.sofa.jraft.rhea.options.RheaKVStoreOptions;
import com.alipay.sofa.jraft.rhea.options.configured.MultiRegionRouteTableOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.PlacementDriverOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.RheaKVStoreOptionsConfigured;
import com.dlg.common.bean.CmdPack;
import com.dlg.common.bus.IBusSender;
import com.dlg.common.bus.MqttBusSender;
import com.dlg.common.checksum.ICheckSum;
import com.dlg.common.codec.IBodyCodec;
import com.dlg.common.codec.IMsgCodec;
import com.dlg.common.hq.MatchData;
import com.dlg.common.uuid.GudyUuid;
import com.dlg.exchange.engin.command.CmdPacketQueue;
import com.dlg.exchange.engin.core.EngineApi;
import com.dlg.exchange.engin.core.EngineCore;
import com.dlg.exchange.engin.db.DbQuery;
import com.dlg.exchange.engin.handler.BaseHandler;
import com.dlg.exchange.engin.handler.match.StockMatchHandler;
import com.dlg.exchange.engin.handler.pub.L1PubHandler;
import com.dlg.exchange.engin.handler.risk.ExistRiskHandler;
import com.dlg.exchange.engin.orderbook.GOrderBookImpl;
import com.dlg.exchange.engin.orderbook.IOrderBook;
import com.dlg.exchange.engin.tools.NetUtils;
import com.google.common.collect.Lists;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.datagram.DatagramSocketOptions;
import lombok.extern.log4j.Log4j2;
import org.eclipse.collections.impl.map.mutable.primitive.IntObjectHashMap;
import org.eclipse.collections.impl.map.mutable.primitive.ShortObjectHashMap;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;

import javax.annotation.PostConstruct;
import java.util.List;

@Log4j2
@SpringBootApplication(exclude= {DruidDataSourceAutoConfigure.class, DataSourceAutoConfiguration.class})
public class EnginApplication implements CommandLineRunner {

    @Autowired
    DbQuery db;

    @Autowired
    private StockMatchHandler matchHandler;
    L1PubHandler pubHandler;
     @Autowired
    private ExistRiskHandler riskHandler;

    @Autowired
    private Vertx vertx;

    @Autowired
    private IBodyCodec bodyCodec;

    @Value("${multicast.sequrllist}")
    private String seqUrlList;
    @Value("${multicast.orderrecvport}")
    private String orderRecvPort;
    @Value("${multicast.orderrecvip}")
    private String orderRecvIp;

    @Autowired
    private IMsgCodec msgCodec;

    private IBusSender busSender;
    @Value("${hqpub.pubip}")
    private String pubIp;
    @Value("${hqpub.pubport}")
    private  String pubPort;
    @Autowired
    private ShortObjectHashMap<List<MatchData>> matcherEventMap;
    @Autowired
    private ICheckSum cs;
    @Value("${engine.id}")
    private String id;
//    @PostConstruct
//private void init(){
//    busSender=  new MqttBusSender(pubip, Integer.parseInt(pubport), msgCodec, vertx);
//}

    public static void main(String[] args) {
        SpringApplication.run(EnginApplication.class, args);
    }


    @Override
    public void run(String... args) throws Exception {
        //启动撮合核心

        //1.前置风控处理器
//        final BaseHandler riskHandler = new ExistRiskHandler(
//                db.queryAllBalance().keySet(),
//                db.queryAllStockCode()
//        );


        //2.撮合处理器(订单簿*****) 撮合/提供行情查询
//        matchHandler.setOrderBookMap(orderBookMap);

        //3.发布处理器
//        ShortObjectHashMap<List<MatchData>> matcherEventMap = new ShortObjectHashMap<>();
//        for (short id : db.queryAllMemberIds()) {
//            matcherEventMap.put(id, Lists.newArrayList());
//        }
//        pubHandler.setMatcherEventMap(matcherEventMap);
        busSender = new MqttBusSender(pubIp, Integer.parseInt(pubPort), msgCodec, vertx);
        busSender.startup();

        final L1PubHandler pubHandler = new L1PubHandler();
        pubHandler.setMatcherEventMap(matcherEventMap);
        pubHandler.setCs(cs);
        pubHandler.setBodyCodec(bodyCodec);
        pubHandler.setId(id);
        pubHandler.setBusSender(busSender);
        EngineApi engineApi = new EngineCore(
                riskHandler,
                matchHandler,
                pubHandler
        ).getApi();

        //4.建立总线连接 初始化数据的发送
//      initPub();



        //5.初始化接收排队机数据以及连接
        final List<RegionRouteTableOptions> regionRouteTableOptions
                = MultiRegionRouteTableOptionsConfigured
                .newConfigured()
                .withInitialServerList(-1L, seqUrlList)
                .config();

        final PlacementDriverOptions pdOpts = PlacementDriverOptionsConfigured
                .newConfigured()
                .withFake(true)
                .withRegionRouteTableOptionsList(regionRouteTableOptions)
                .config();

        final RheaKVStoreOptions opts = RheaKVStoreOptionsConfigured
                .newConfigured()
                .withPlacementDriverOptions(pdOpts)
                .config();

         final RheaKVStore orderKvStore = new DefaultRheaKVStore();

        orderKvStore.init(opts);

        /////////////////////////////////////////////////////////////////////

        //委托指令处理器
        CmdPacketQueue.getInstance().init(orderKvStore, bodyCodec, engineApi);

        //组播 允许多个Socket接收同一份数据
        DatagramSocket socket = vertx.createDatagramSocket(new DatagramSocketOptions());
        socket.listen(Integer.parseInt(orderRecvPort), "0.0.0.0", asyncRes -> {
            if (asyncRes.succeeded()) {

                socket.handler(packet -> {
                    Buffer udpData = packet.data();
                    if (udpData.length() > 0) {
                        try {
                            CmdPack cmdPack = bodyCodec.deserialize(udpData.getBytes(), CmdPack.class);
                            CmdPacketQueue.getInstance().cache(cmdPack);
                        } catch (CodecException e) {
                            log.error("decode packet error", e);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    } else {
                        log.error("recv empty udp packet from client : {}", packet.sender().toString());
                    }
                });


                try {
                    socket.listenMulticastGroup(orderRecvIp,
                            NetUtils.mainInterface().getName(), null, asyncRes2 -> {
                                log.info("listen succeed {}", asyncRes2.succeeded());
                            });
                } catch (Exception e) {
                    log.error(e);
                }
            } else {
                log.error(" Listen failed ,", asyncRes.cause());
            }
        });


    }


}
